package com.shujia.indicators.usertag;

import com.shujia.indicators.utils.SparkUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;

public class UserTagEtl {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkUtils.initSession();

        Dataset<Row> memberBase = sparkSession.sql(
                "select id as memberId,phone,sex,member_channel as channel,mp_open_id as subOpenId," +
                        " address_default as address_default,date_format(create_time,'yyyy-MM-dd') as regTime " +
                        " from t_member"
        );

        Dataset<Row> orderBehavior = sparkSession.sql(
                "select o.member_id as memberId, count(o.order_id) as orderCount, " +
                        " date_format( max(o.create_time),'yyyy-MM-dd') as orderTime, " +
                        " sum(o.pay_price) as orderMoney, " +
                        " collect_list(distinct oc.commodity_id) as favGoods" +
                        " from t_order as o left join t_order_commodity as oc on o.order_id = oc.order_id group by o.member_id "
        );

        Dataset<Row> freeCount = sparkSession.sql(
                "select member_id as memberId, date_format(create_time,'yyyy-MM-dd') as freeCouponTime" +
                        " from t_coupon_member where coupon_id = 1 "
        );
        Dataset<Row> couponTimes = sparkSession.sql(
                "select member_id as memberId, collect_list(date_format(create_time,'yyyy-MM-dd')) as couponTimes " +
                        " from t_coupon_member where coupon_id <> 1 group by member_id"
        );
        Dataset<Row> chargeMoney = sparkSession.sql(
                "select cm.member_id as memberId, sum(c.coupon_price/2) as chargePrice " +
                        " from t_coupon_member as cm left join t_coupon as c " +
                        "on cm.coupon_id = c.id where cm.coupon_channel = 1 group by cm.member_id"
        );
        Dataset<Row> overTime = sparkSession.sql(
                "select member_id as memberId, (to_unix_time(max(arrive_time)) - to_unix_time(max(pick_time))) as overTime " +
                        " from t_delivery group by member_id"
        );
        Dataset<Row> feedBack = sparkSession.sql(
                "select fb.member_id as member_id,fb.feedback_type as feedBack " +
                        "from t_feedback as fb " +
                        " right join (select max(id) as fid from t_feedback group by member_id ) as t " +
                        " on fb.id = t.fid"
        );

        memberBase.registerTempTable("memberBase");
        orderBehavior.registerTempTable("orderBehavior");
        freeCount.registerTempTable("freeCount");
        couponTimes.registerTempTable("couponTimes");
        chargeMoney.registerTempTable("chargeMoney");
        overTime.registerTempTable("overTime");
        feedBack.registerTempTable("feedBack");

        Dataset<Row> result = sparkSession.sql(
                "select m.* , o.orderCount, o.orderTime, o.orderMoney, o.favGoods, " +
                        " fc.freeCouponTime, ct.couponTimes, cm.chargePrice, ot.overTime, fb.feedBack " +
                        " from memberBase as m " +
                        " left join orderBehavior as o on m.memberId = o.memberId" +
                        " left join freeCount  as fc on m.memberId = fc.memberId" +
                        " left join couponTimes as ct on m.memberId = ct.memberId" +
                        " left join chargeMoney as cm on m.memberId = cm.memberId" +
                        " left join overTime as ot on m.memberId = ot.memberId" +
                        " left join feedBack as fb on m.memberId = fb.memberId"
        );


        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost:2181");
        Connection conn = null;
        Admin admin = null;
        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
//      创建表名
            TableName userTags = TableName.valueOf("UserTags");
            //创建表描述器
            if (conn.getTable(userTags)==null){
                TableDescriptorBuilder descriptorBuilder = TableDescriptorBuilder.newBuilder(userTags);

                ColumnFamilyDescriptor memberBaseCol = ColumnFamilyDescriptorBuilder.of("memberBase");

                ColumnFamilyDescriptor userBehavior = ColumnFamilyDescriptorBuilder.of("UserBehavior");

                ColumnFamilyDescriptor spending = ColumnFamilyDescriptorBuilder.of("spending");

                ColumnFamilyDescriptor service = ColumnFamilyDescriptorBuilder.of("Service");

                TableDescriptorBuilder userTagsTable = descriptorBuilder.setColumnFamily(memberBaseCol)
                        .setColumnFamily(userBehavior)
                        .setColumnFamily(spending)
                        .setColumnFamily(service);

                admin.createTable(userTagsTable.build());
            }


            Table table = conn.getTable(userTags);
            int count = 1;
            for (Row row : result.collectAsList()) {
                ArrayList<Put> puts = new ArrayList<>();
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("memberBase"), Bytes.toBytes("memberId"), Bytes.toBytes((int) row.get(1))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("memberBase"), Bytes.toBytes("phone"), Bytes.toBytes((String) row.get(2))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("memberBase"), Bytes.toBytes("sex"), Bytes.toBytes((String) row.get(3))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("memberBase"), Bytes.toBytes("channel"), Bytes.toBytes((String) row.get(4))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("memberBase"), Bytes.toBytes("subOpenId"), Bytes.toBytes((String) row.get(5))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("UserBehavior"), Bytes.toBytes("orderCount"), Bytes.toBytes((Integer) row.get(6))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("UserBehavior"), Bytes.toBytes("orderTime"), Bytes.toBytes(String.valueOf(row.get(7)))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("UserBehavior"), Bytes.toBytes("orderMoney"), Bytes.toBytes((BigDecimal) row.get(8))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("UserBehavior"), Bytes.toBytes("favGoods"), Bytes.toBytes((String) row.get(9))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("spending"), Bytes.toBytes("freeCouponTime"), Bytes.toBytes((String) row.get(10))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("spending"), Bytes.toBytes("couponTimes"), Bytes.toBytes((String) row.get(11))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("spending"), Bytes.toBytes("chargePrice"), Bytes.toBytes((String) row.get(12))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("service"), Bytes.toBytes("overTime"), Bytes.toBytes((String) row.get(13))));
                puts.add(new Put(Bytes.toBytes(count)).addColumn(Bytes.toBytes("service"), Bytes.toBytes("feedBack"), Bytes.toBytes((String) row.get(14))));

                table.put(puts);
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                if (admin != null) {
                    admin.close();
                }
                if (conn != null) {
                    conn.close();
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

    }
}
